package org.spiderflow.core;

import com.alibaba.ttl.TtlRunnable;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.spiderflow.common.concurrent.SpiderFlowThreadPoolExecutor;
import org.spiderflow.common.concurrent.SubThreadPoolExecutor;
import org.spiderflow.common.concurrent.ThreadSubmitStrategy;
import org.spiderflow.common.concurrent.ThreadSubmitStrategyFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spiderflow.common.context.SpiderContext;
import org.spiderflow.common.context.SpiderContextHolder;
import org.spiderflow.core.executor.shape.LoopExecutor;
import org.spiderflow.core.model.FlowStatus;
import org.spiderflow.core.model.LoopParam;
import org.spiderflow.core.model.SpiderFlow;
import org.spiderflow.core.service.FlowNoticeService;
import org.spiderflow.core.utils.ExecutorsUtils;
import org.spiderflow.core.utils.ExpressionUtils;
import org.spiderflow.core.utils.SpiderFlowUtils;
import org.spiderflow.common.enums.FlowNoticeType;
import org.spiderflow.common.executor.ShapeExecutor;
import org.spiderflow.common.listener.SpiderListener;
import org.spiderflow.common.model.SpiderNode;
import org.spiderflow.common.model.SpiderOutput;
import org.spiderflow.common.model.SpiderTask;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.lang.reflect.Array;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 爬虫的核心类
 *
 * @author jmxd
 */
@Component
public class Spider {
    private static final Logger LOGGER = LoggerFactory.getLogger(Spider.class);
    private final Optional<List<SpiderListener>> listeners;

    @Value("${spider.thread.max:64}")
    private Integer totalThreads;

    @Value("${spider.thread.default:8}")
    private Integer defaultThreads;

    @Value("${spider.detect.dead-cycle:5000}")
    private Integer deadCycle;

    private final FlowNoticeService flowNoticeService;

    private static SpiderFlowThreadPoolExecutor executorInstance;

    private static final String ATOMIC_DEAD_CYCLE = "__atomic_dead_cycle";


    public Spider(Optional<List<SpiderListener>> listeners, FlowNoticeService flowNoticeService) {
        this.listeners = listeners;
        this.flowNoticeService = flowNoticeService;
    }

    @PostConstruct
    private void init() {
        setExecutorInstance(new SpiderFlowThreadPoolExecutor(totalThreads));
    }

    public static SpiderFlowThreadPoolExecutor getExecutorInstance() {
        return executorInstance;
    }

    public static void setExecutorInstance(SpiderFlowThreadPoolExecutor executorInstance) {
        Spider.executorInstance = executorInstance;
    }

    public List<SpiderOutput> run(SpiderFlow spiderFlow, SpiderContext context, Map<String, Object> variables) {
        Map<String, Object> variablesMap = new HashMap<>(8);
        if (!CollectionUtils.isEmpty(variables)) {
            variablesMap.putAll(variables);
        }
        SpiderNode root = SpiderFlowUtils.loadXMLFromString(spiderFlow.getXml());
        // 流程开始通知
        flowNoticeService.sendFlowNotice(spiderFlow, FlowNoticeType.START_NOTICE);
        executeRoot(root, context, variablesMap);
        // 流程结束通知
        flowNoticeService.sendFlowNotice(spiderFlow, FlowNoticeType.END_NOTICE);
        return context.getOutputs();
    }

    public void run(SpiderFlow spiderFlow, SpiderContext context) {
        run(spiderFlow, context, null);
    }

    public void runWithTest(SpiderNode root, SpiderContext context) {
        //将上下文存到ThreadLocal里，以便后续使用
        SpiderContextHolder.set(context);
        //死循环检测的计数器（死循环检测只在测试时有效）
        AtomicInteger executeCount = new AtomicInteger(0);
        //存入到上下文中，以供后续检测
        context.put(ATOMIC_DEAD_CYCLE, executeCount);
        //执行根节点
        executeRoot(root, context, new HashMap<>());
        //当爬虫任务执行完毕时,判断是否超过预期
        if (executeCount.get() > deadCycle) {
            LOGGER.error("检测到可能出现死循环,测试终止");
        } else {
            LOGGER.info("测试完毕！");
        }
        //将上下文从ThreadLocal移除，防止内存泄漏
        SpiderContextHolder.remove();
    }

    /**
     * 执行根节点
     */
    private void executeRoot(SpiderNode root, SpiderContext context, Map<String, Object> variables) {
        //构建提交策略
        String strategy = root.getStringJsonValue("submit-strategy");
        ThreadSubmitStrategy submitStrategy = ThreadSubmitStrategyFactory.createStrategy(strategy);
        //获取当前流程执行线程数
        int nThreads = NumberUtils.toInt(root.getStringJsonValue(ShapeExecutor.THREAD_COUNT), defaultThreads);
        //创建子线程池，采用一父多子的线程池,子线程数不能超过总线程数（超过时进入队列等待）,+1是因为会占用一个线程用来调度执行下一级
        SubThreadPoolExecutor pool = executorInstance.createSubThreadPoolExecutor(Math.max(nThreads, 1) + 1, submitStrategy);
        context.setRootNode(root);
        context.setThreadPool(pool);
        //触发监听器
        listeners.ifPresent(listenerList -> listenerList.forEach(listener -> listener.beforeStart(context)));
        //启动一个线程开始执行任务,并监听其结束并执行下一级
        Future<Object> f = pool.submitAsync(TtlRunnable.get(() -> {
            try {
                //执行具体节点
                Spider.this.executeNode(null, root, context, variables);
                Queue<Future<SpiderTask>> queue = context.getFutureQueue();
                //循环从队列中获取Future,直到队列为空结束,当任务完成时，则执行下一级
                while (!queue.isEmpty()) {
                    executeQueueTask(context, submitStrategy.comparator(), queue);
                }
                //等待线程池结束
                pool.awaitTermination();
            } finally {
                //触发监听器
                listeners.ifPresent(listenerList -> listenerList.forEach(listener -> listener.afterEnd(context)));
            }
        }), Object.class, root);
        try {
            f.get();    //阻塞等待所有任务执行完毕
        } catch (ExecutionException e) {
            //do nothing
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void executeQueueTask(SpiderContext context, Comparator<SpiderNode> comparator,
                                  Queue<Future<SpiderTask>> queue) {
        //判断任务是否完成
        getEarliestCompletedTask(comparator, queue).ifPresent(max -> {
            queue.remove(max);
            //检测是否运行中(当在页面中点击"停止"时,此值为false,其余为true)
            if (!context.isRunning()) {
                return;
            }
            try {
                SpiderTask task = max.get();
                SpiderNode taskNode = task.getNode();
                taskNode.decrement();    //任务执行完毕,计数器减一(该计数器是给Join节点使用)
                //判断是否允许执行下一级
                if (task.getExecutor().allowExecuteNext(taskNode, context, task.getVariables())) {
                    LOGGER.debug("执行节点[{}:{}]完毕", taskNode.getNodeName(), taskNode.getNodeId());
                    //执行下一级
                    Spider.this.executeNextNodes(taskNode, context, task.getVariables());
                } else {
                    LOGGER.debug("执行节点[{}:{}]完毕，忽略执行下一节点", taskNode.getNodeName(), taskNode.getNodeId());
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                LOGGER.error("程序发生异常", e);
            } catch (ExecutionException e) {
                LOGGER.error("程序发生异常", e);
            }
        });
        try {
            //睡眠1ms,让出cpu
            Thread.sleep(1);
        } catch (InterruptedException ignored) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 取出最早完成的任务
     *
     * @param comparator 比较器
     * @param queue      队列
     * @return 最早完成的任务
     */
    private Optional<Future<SpiderTask>> getEarliestCompletedTask(Comparator<SpiderNode> comparator, Queue<Future<SpiderTask>> queue) {
        return queue.stream().filter(Future::isDone).max((o1, o2) -> {
            try {
                return comparator.compare(o1.get().getNode(), o2.get().getNode());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (ExecutionException e) {
                //do nothing
            }
            return 0;
        });
    }

    /**
     * 执行下一级节点
     */
    private void executeNextNodes(SpiderNode node, SpiderContext context, Map<String, Object> variables) {
        List<SpiderNode> nextNodes = node.getNextNodes();
        if (nextNodes != null) {
            for (SpiderNode nextNode : nextNodes) {
                executeNode(node, nextNode, context, variables);
            }
        }
    }

    /**
     * 执行节点
     */
    public void executeNode(SpiderNode fromNode, SpiderNode node, SpiderContext context, Map<String, Object> variables) {
        String shape = node.getStringJsonValue("shape");
        if (StringUtils.isBlank(shape)) {
            executeNextNodes(node, context, variables);
            return;
        }
        //判断箭头上的条件，如果不成立则不执行
        if (!executeCondition(fromNode, node, variables)) {
            return;
        }
        LOGGER.debug("执行节点[{}:{}]", node.getNodeName(), node.getNodeId());
        //找到对应的执行器
        Optional<ShapeExecutor> executor = ExecutorsUtils.get(shape);
        if (!executor.isPresent()) {
            LOGGER.error("执行失败,找不到对应的执行器:{}", shape);
            context.setRunning(false);
            return;
        }
        String loopCountStr = node.getStringJsonValue(ShapeExecutor.LOOP_COUNT);
        boolean isLoop = StringUtils.isNotBlank(loopCountStr);
        LoopParam param = buildLoopParam(node, variables, loopCountStr, isLoop);
        if (param.getLoopCount() <= 0) {
            return;
        }
        //获取循环下标的变量名称
        List<SpiderTask> tasks = buildTasks(fromNode, node, context, variables, executor.get(), param);
        runTask(node, context, executor.get(), tasks);
    }

    private List<SpiderTask> buildTasks(SpiderNode fromNode, SpiderNode node, SpiderContext context,
                                        Map<String, Object> variables, ShapeExecutor executor, LoopParam param) {
        String loopVariableName = node.getStringJsonValue(ShapeExecutor.LOOP_VARIABLE_NAME);
        String loopItem = node.getStringJsonValue(LoopExecutor.LOOP_ITEM, "item");
        List<SpiderTask> tasks = new ArrayList<>();
        for (int i = param.getLoopStart(); i < param.getLoopEnd(); i++) {
            node.increment();    //节点执行次数+1(后续Join节点使用)
            if (!context.isRunning()) {
                continue;
            }
            Map<String, Object> nVariables = buildVariables(fromNode, node, variables, param, loopVariableName, loopItem, i);
            tasks.add(newTask(node, context, executor, nVariables));
        }
        return tasks;
    }

    private Map<String, Object> buildVariables(SpiderNode fromNode, SpiderNode node, Map<String, Object> variables,
                                               LoopParam param, String loopVariableName, String loopItem, int i) {
        Map<String, Object> nVariables = new HashMap<>(8);
        // 判断是否需要传递变量
        if (fromNode == null || node.isTransmitVariable(fromNode.getNodeId())) {
            nVariables.putAll(variables);
        }
        if (!param.isLoop()) {
            return nVariables;
        }
        // 存入下标变量
        if (!StringUtils.isBlank(loopVariableName)) {
            nVariables.put(loopVariableName, i);
        }
        // 存入item
        Object item = param.getLoopArray() == null ? i : Array.get(param.getLoopArray(), i);
        nVariables.put(loopItem, item);
        return nVariables;
    }

    private SpiderTask newTask(SpiderNode node, SpiderContext context, ShapeExecutor executor, Map<String, Object> nVariables) {
        return new SpiderTask(TtlRunnable.get(() -> {
            if (!context.isRunning()) {
                return;
            }
            try {
                //死循环检测，当执行节点次数大于阈值时，结束本次测试
                AtomicInteger executeCount = context.get(ATOMIC_DEAD_CYCLE);
                if (executeCount != null && executeCount.incrementAndGet() > deadCycle) {
                    context.setRunning(false);
                    return;
                }
                //执行节点具体逻辑
                executor.execute(node, context, nVariables);
                //当未发生异常时，移除ex变量
                nVariables.remove("ex");
            } catch (RuntimeException t) {
                nVariables.put("ex", t);
                LOGGER.error("执行节点[{}:{}]出错,异常信息：{}", node.getNodeName(), node.getNodeId(), t);
            }
        }), node, nVariables, executor);
    }

    private void runTask(SpiderNode node, SpiderContext context, ShapeExecutor executor, List<SpiderTask> tasks) {
        LinkedBlockingQueue<Future<SpiderTask>> futureQueue = context.getFutureQueue();
        for (SpiderTask task : tasks) {
            //判断节点是否是异步运行
            if (executor.isThread()) {
                //提交任务至线程池中,并将Future添加到队列末尾
                futureQueue.add(context.getThreadPool().submitAsync(task.getRunnable(), task, node));
            } else {
                FutureTask<SpiderTask> futureTask = new FutureTask<>(task.getRunnable(), task);
                futureTask.run();
                futureQueue.add(futureTask);
            }
        }
    }

    private static LoopParam buildLoopParam(SpiderNode node, Map<String, Object> variables, String loopCountStr, boolean isLoop) {
        LoopParam param = new LoopParam(isLoop);
        if (isLoop) {
            try {
                Object loopArray = ExpressionUtils.execute(loopCountStr, variables);
                if (loopArray == null) {
                    param.setLoopCount(0);
                } else if (loopArray instanceof Collection<?>) {
                    param.setLoopCount(((Collection<?>) loopArray).size())
                         .setLoopArray(((Collection<?>) loopArray).toArray());
                } else if (loopArray.getClass().isArray()) {
                    param.setLoopCount(Array.getLength(loopArray))
                         .setLoopArray(loopArray);
                } else {
                    param.setLoopCount(NumberUtils.toInt(loopArray.toString(), 0));
                }
                param.setLoopEnd(param.getLoopCount());
                if (param.getLoopCount() > 0) {
                    param.setLoopStart(Math.max(NumberUtils.toInt(node.getStringJsonValue(LoopExecutor.LOOP_START), 0), 0));
                    int end = NumberUtils.toInt(node.getStringJsonValue(LoopExecutor.LOOP_END), -1);
                    if (end >= 0) {
                        param.setLoopEnd(Math.min(end, param.getLoopEnd()));
                    } else {
                        param.setLoopEnd(Math.max(param.getLoopEnd() + end + 1, 0));
                    }
                }
                LOGGER.info("获取循环参数{}={}", loopCountStr, param);
            } catch (RuntimeException e) {
                param.setLoopCount(0);
                LOGGER.error("获取循环参数失败,loopCountStr->{},异常信息：{}", loopCountStr, e.getMessage(), e);
            }
        }
        return param;
    }

    /**
     * 判断箭头上的表达式是否成立
     */
    private boolean executeCondition(SpiderNode fromNode, SpiderNode node, Map<String, Object> variables) {
        if (fromNode == null) {
            return true;
        }
        boolean hasException = variables.get("ex") != null;
        String exceptionFlow = node.getExceptionFlow(fromNode.getNodeId());
        //当出现异常流转 : 1
        boolean errorButNoException = FlowStatus.ERROR.code().equalsIgnoreCase(exceptionFlow) && !hasException;
        //未出现异常流转 : 2
        boolean normalButHasException = FlowStatus.NORMAL.code().equalsIgnoreCase(exceptionFlow) && hasException;
        if (errorButNoException || normalButHasException) {
            return false;
        }
        String condition = node.getCondition(fromNode.getNodeId());
        // 判断是否有条件
        if (StringUtils.isBlank(condition)) {
            return true;
        }
        try {
            Object result = ExpressionUtils.execute(condition, variables);
            boolean isContinue = Objects.equals(Boolean.TRUE.toString(), result) || Objects.equals(Boolean.TRUE, result);
            LOGGER.debug("判断{}={}", condition, isContinue);
            return isContinue;
        } catch (Exception e) {
            LOGGER.error("判断{}出错,异常信息：{}", condition, e);
            return false;
        }
    }
}
